var config = require('./config/index.js');

var kafka = require('kafka-node');
var Producer = kafka.Producer;

var client = new kafka.Client(config.zookeeper.uri);
var producer = new Producer(client, {
    requireAcks: 1,
    ackTimeoutMs: 1000
});

var __topic = 't1';

producer.on('error', function(err) {
    console.log('error', err);
});

producer.on('ready', function() {
    console.log('ready');

    // 
    producer.createTopics([__topic], false, function(err, data) {
        if (err) throw err;
        console.log(data);

        // ---producer send msg
        i = 0;
        setInterval(function() {
            i++;
            producer.send([{
                topic: __topic,
                messages: 'hi ' + i,
                partition: 0
            }], function(err, data) {
                if (err) return console.log('send msg err', err);
                console.log('send msg ok', data);
            });
        }, 2000);
    });
});

// bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka --from-beginning --topic t1